package com.tjhk.kafka.demo;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.util.Collections;
import java.util.Properties;

/**
 * @ClassName AclConsumer
 * @Description 模拟acl consumer
 * @Author yang
 * @Date 2020/8/1
 * @Version 1.0
 **/
public class AclConsumer {
    private final static Logger LOGGER = LoggerFactory.getLogger(AclConsumer.class);

    public static void main(String[] args) {
        String topic = "test-topic";
        KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(loadProp());
        consumer.subscribe(Collections.singleton(topic));
        while (true) {
            ConsumerRecords<String, byte[]> records = consumer.poll(500);
            for (ConsumerRecord<String, byte[]> record : records) {

                try {
                    System.out.printf("partition= %d, offset = %d, key = %s, value = %s\n", record.partition(), record.offset(), record.key(), new String(record.value(),"UTF-8"));
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
            }
            consumer.commitAsync();
        }
    }

    private static Properties loadProp(){
        final Properties prop = new Properties();
        prop.put("bootstrap.servers","192.168.244.151:9092,192.168.244.152:9092,192.168.244.153:9092");
        prop.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        prop.put("value.deserializer", "org.apache.kafka.common.serialization.ByteArrayDeserializer");
        //consumer-jaas.properties
        prop.put("security.protocol", "SASL_PLAINTEXT");
        prop.put("sasl.mechanism", "SCRAM-SHA-512");
        prop.put("group.id", "test-group");

        // 可选设置属性
        prop.put("enable.auto.commit", "true");
        // 自动提交offset,每1s提交一次
        prop.put("auto.commit.interval.ms", "10000");
        prop.put("auto.offset.reset", "earliest");
        prop.put("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username='test-consumer' password='123456';");
        return prop;
    }
}
